package exlocalsubpub

import (
	"fmt"
	"gitee.com/ymofen/gobase"
	"gitee.com/ymofen/gobase/factory"
	"gitee.com/ymofen/gobase/gotask"
	"gitee.com/ymofen/gobase/gotask/chantask"
	"gitee.com/ymofen/gobase/subpub"
	"io"
	"sync/atomic"
)

type SubExLocalSession struct {
	closedflag int32
	notify     subpub.ISubPub
	sessionid  string
	conf       gobase.StrMap
	topic      string

	pubque     string
	pubqueTask *chantask.ChanTask
	pubfun     func(topic string, args ...interface{}) int

	copybufflag int8

	operaRef *gobase.ObjectInnerRef
	opera    *ExLocalNode
}

func CreateSubExLocalSession(args ...interface{}) (rval interface{}, err error) {
	plg := &SubExLocalSession{
		sessionid: args[0].(string),
		conf:      args[1].(gobase.StrMap),
		notify:    args[2].(subpub.ISubPub),
	}
	return plg, nil
}

func (this *SubExLocalSession) Open() error {
	if gobase.GoFunCatchException {
		defer gobase.DeferCatchPanic()
	}
	gotask.DefaultPatrolTask().CheckStart()
	return nil
}

func (this *SubExLocalSession) PostData(args ...interface{}) error {
	return nil
}

func (this *SubExLocalSession) innerPubQueue(topic string, args ...interface{}) int {
	pubtask := this.pubqueTask
	if pubtask != nil {
		pubtask.PushTask(func(args ...interface{}) {
			if atomic.LoadInt32(&this.closedflag) == 1 {
				return
			}
			this.innerPub(topic, args...)
		}, args...)
		return 1
	} else {
		return 0
	}
}

func (this *SubExLocalSession) innerPub(topic string, args ...interface{}) int {
	ex := this.notify
	if ex == nil {
		return 0
	}
	ex.Pub(this.sessionid, 0, args...)
	return 1
}

// 连接参数 conntype=exlocal&extype=0
func (this *SubExLocalSession) Start() {

	connstr := this.conf.URLFormEncode()

	this.operaRef = DefaultExGroup.CheckGetRef(connstr)
	this.opera = this.operaRef.GetObject().(*ExLocalNode)
	this.opera.checkStart()

	topic := this.conf.StringByName("topic", "")
	this.topic = topic
	if len(this.topic) == 0 {
		this.notify.Pub(this.sessionid, subpub.ACTION_CONNECTFAIL, fmt.Errorf("未指定topic参数"))
		return
	}

	this.pubque = this.conf.StringByName("pubque", "")
	if len(this.pubque) > 0 {
		this.pubqueTask, _ = chantask.DefaultChanTaskHub.CheckGetChanTask(this.pubque, 1024, 0)
		this.pubfun = this.innerPubQueue
	} else {
		this.pubfun = this.innerPub
	}

	this.opera.ex.Sub(this.sessionid, topic, func(id, topic string, args ...interface{}) (ok bool) {
		if len(args) == 0 {
			return false
		}
		if str, ok := args[0].(string); ok {
			this.pubfun(this.sessionid, subpub.ACTION_RECV_TOPIC_DATA, topic, []byte(str))
			return true
		}
		if buf, ok := args[0].([]byte); ok {
			this.pubfun(this.sessionid, subpub.ACTION_RECV_TOPIC_DATA, topic, buf)
			return true
		}

		return true
	})
}

func (this *SubExLocalSession) Close() error {
	if !atomic.CompareAndSwapInt32(&this.closedflag, 0, 1) {
		return io.ErrClosedPipe
	}
	if this.operaRef != nil {
		this.opera.ex.Unsub(this.sessionid, "evt")
		this.opera.ex.Unsub(this.sessionid, this.topic)
		this.opera = nil

		DefaultExGroup.ReleaseRef(this.operaRef)
		this.operaRef = nil
	}

	gotask.DefaultPatrolTask().DelTask(this.sessionid)

	this.notify.Pub(this.sessionid, 0, subpub.ACTION_CLOSE)
	this.notify = nil
	return nil
}

func init() {
	factory.RegisterCreateFunc("sub.exlocal", CreateSubExLocalSession)
}
